package com.ziroom.finance.rabbitmq.consumer;


import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.QueueingConsumer;
import com.ziroom.finance.rabbitmq.common.EndPoint;
import org.apache.commons.lang.SerializationUtils;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;


/**
 * 读取队列的程序端，实现了Runnable接口。
 *
 * @author syntx
 */
public class QueueConsumer extends EndPoint implements Runnable, Consumer {

    public QueueConsumer(String endPointName) throws IOException, TimeoutException {
        super(endPointName);
    }

    public void run() {
        try {
            //监听，手动返回
            channel.basicConsume(endPointName, false, this);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * Called when consumer is registered.
     */
    public void handleConsumeOk(String consumerTag) {
        System.out.println("Consumer " + consumerTag + " registered");
    }

    /**
     * Called when new message is available.
     */
    public void handleDelivery(String consumerTag, Envelope env,
                               BasicProperties props, byte[] body) throws IOException {
        Map map = (HashMap) SerializationUtils.deserialize(body);

        System.out.println(Thread.currentThread().getName() + " Message Number " + map.get("message number") + " received.");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println(" [x] Done");
            //返回确认信息
            channel.basicAck(env.getDeliveryTag(), false);
        }
    }

    public void handleCancel(String consumerTag) {
        System.out.println(consumerTag + " handleCancel");
    }

    public void handleCancelOk(String consumerTag) {
        System.out.println(consumerTag + " handleCancelOk");
    }

    public void handleRecoverOk(String consumerTag) {
        System.out.println(consumerTag + " handleCancelOk");

    }

    public void handleShutdownSignal(String consumerTag, ShutdownSignalException arg1) {
        System.out.println(consumerTag + " handleShutdownSignal");
    }
}